package com.guqj.thread.concurrent.lock;

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerComsumerTest {

	public static void main(String[] args) {
		MyBlockingQueue<String> queue = new MyBlockingQueue<>(10);
		Producer producer = new Producer(queue);
		Comsumer comsumer = new Comsumer(queue);
		producer.start();
		comsumer.start();
		producer.interrupt();
	}
}

/**
 * lock()/unlock()：就是普通的获取锁和释放锁方法，lock()会阻塞直到成功。
 * lockInterruptibly()：与lock()的不同是，它可以响应中断，如果被其他线程中断了，抛出InterruptedException。
 * @author guqj
 * @date   2017年3月14日
 * @param <E>
 */
class MyBlockingQueue<E> {
	private Queue<E> queue = null;
	private int limit = 10;
	private Lock lock = new ReentrantLock();
	private Condition notFull = lock.newCondition();
	private Condition notEmpty = lock.newCondition();

	public MyBlockingQueue(int limit) {
		this.limit = limit;
		queue = new ArrayBlockingQueue<>(limit);
	}

	public E take() throws InterruptedException {
		lock.lockInterruptibly(); // 优先考虑响应中断，而不是响应锁定的普通获取或重入获取
		try {
			while (queue.isEmpty()) {
				notEmpty.await();
			}
			E e = queue.poll();
			notFull.signal();
			return e;
		} finally {
			lock.unlock();
		}
	}

	public void put(E e) throws InterruptedException {
		lock.lockInterruptibly();
		try {
			while (this.queue.size() == limit) {
				notFull.await();
			}
			queue.add(e);
			notEmpty.signal();
		} finally {
			lock.unlock();
		}
	}
}

class Producer extends Thread {

	private MyBlockingQueue<String> queue;

	public Producer(MyBlockingQueue<String> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {
		int num = 0;
		try {
			while (true) {
				String task = String.valueOf(num);
				queue.put(task);
				System.out.println("produce task " + task);
				num++;
				Thread.sleep((int) (Math.random() * 100));
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

class Comsumer extends Thread {
	private MyBlockingQueue<String> queue;

	public Comsumer(MyBlockingQueue<String> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {
		try {
			while (true) {
				String task = queue.take();
				System.out.println("handle task " + task);
				Thread.sleep((int) (Math.random() * 100));
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}